package com.yomob.kylin.client;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections.CollectionUtils;
import org.bson.Document;
import org.bson.conversions.Bson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateManyModel;
import com.mongodb.client.model.UpdateOptions;
import com.yomob.kylin.entity.Result;

public class MongoDBClient {
    private static final Logger LOGGER = LoggerFactory.getLogger(MongoDBClient.class);

    private MongoDatabase mongoDatabase;

    public MongoDBClient(MongoDatabase mongoDatabase) {
        this.mongoDatabase = mongoDatabase;
    }

    public MongoDatabase getMongoDatabase() {
        return mongoDatabase;
    }

    public MongoCollection<Document> getMongoCollection(String collectionName) {
        return getMongoDatabase().getCollection(collectionName);
    }

    public void saveToMongo(List<Result> results, String collectionName) {
        for (Result result : results) {
            saveToMongo(result.getMatches(), result.getValues(), collectionName);
        }
    }

    public void saveToMongo(Map<String, Object> matches, Map<String, Object> values, String collectionName) {
        List<Bson> eqs = Lists.newArrayListWithCapacity(matches.size());
        matches.forEach((s, o) -> {
            Bson eq = Filters.eq(s, o);
            eqs.add(eq);
        });
        Bson filter = Filters.and(eqs);
        Document update = new Document();
        values.forEach((s, o) -> update.append("$set", new Document(s, o)));
        saveToMongo(filter, update, collectionName);
    }

    public void saveToMongo(Bson filter, Bson update, String collectionName) {
        UpdateOptions options = new UpdateOptions().upsert(true);
        getMongoCollection(collectionName).updateOne(filter, update, options);
    }
    public void saveToMongoBulkByBatch(List<UpdateManyModel<Document>> models, String collectionName) {
        System.out.println(("Saving result to mongo. Result size: " + models.size()));
        if (CollectionUtils.isNotEmpty(models)) {
            List<List<UpdateManyModel<Document>>> res = split(models, 5000);
            MongoCollection<Document> c = getMongoCollection(collectionName);
            for (List<UpdateManyModel<Document>> mos : res) {
                c.bulkWrite(mos, new BulkWriteOptions().ordered(false));
            }
        }
    }
    public void saveToMongoBulk(List<UpdateManyModel<Document>> models, String collectionName) {
        LOGGER.info("Saving result to mongo. Result size: {}", models.size());
        if (CollectionUtils.isNotEmpty(models)) {
            getMongoCollection(collectionName).bulkWrite(models, new BulkWriteOptions().ordered(false));
        }
    }
    public static  <T> List<List<T>> split(List<T> resList, int count){
        if(resList==null ||count<1)
            return  null ;
        List<List<T>> ret=new ArrayList<List<T>>();
        int size=resList.size();
        if(size<=count){
            ret.add(resList);
        }else{
            int pre=size/count;
            int last=size%count;
            for(int i=0;i<pre;i++){
                List<T> itemList=new ArrayList<T>();
                for(int j=0;j<count;j++){
                    itemList.add(resList.get(i*count+j));
                }
                ret.add(itemList);
            }
            if(last>0){
                List<T> itemList=new ArrayList<T>();
                for(int i=0;i<last;i++){
                    itemList.add(resList.get(pre*count+i));
                }
                ret.add(itemList);
            }
        }
        return ret;
    }
}
